package com.glodon.paas.job.agent.mq;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MultipleDestinationMessageConsumer implements MessageConsumer {
    private static final Logger logger = LoggerFactory.getLogger(MultipleDestinationMessageConsumer.class);

    private Map<String, MessageConsumer> consumers = new HashMap<String, MessageConsumer>();
    private MessageListener listener;

    public MultipleDestinationMessageConsumer(Map<String, MessageConsumer> consumers) {
        this.consumers = consumers;
    }

    public void remove(String type) {
        try {
            MessageConsumer consumer = consumers.remove(type);
            consumer.close();
        } catch (JMSException e) {
            logger.error("remove worker type failure", e);
            throw new RuntimeException(e);
        }
    }

    @Override
    public void setMessageListener(MessageListener listener) throws JMSException {
        this.listener = listener;
        for (MessageConsumer consumer : consumers.values()) {
            consumer.setMessageListener(listener);
        }
    }

    @Override
    public Message receiveNoWait() throws JMSException {
        for (MessageConsumer consumer : consumers.values()) {
            Message m = consumer.receiveNoWait();
            if (m != null)
                return m;
        }
        return null;
    }

    @Override
    public Message receive(long timeout) throws JMSException {
        for (MessageConsumer consumer : consumers.values()) {
            Message m = consumer.receive(timeout / consumers.size());
            if (m != null)
                return m;
        }
        return null;
    }

    @Override
    public Message receive() throws JMSException {
        for (MessageConsumer consumer : consumers.values()) {
            Message m = consumer.receive();
            if (m != null)
                return m;
        }
        return null;
    }

    @Override
    public String getMessageSelector() throws JMSException {
        return consumers.values().iterator().next().getMessageSelector();
    }

    @Override
    public MessageListener getMessageListener() throws JMSException {
        return listener;
    }

    @Override
    public void close() throws JMSException {
        for (MessageConsumer consumer : consumers.values()) {
            consumer.close();
        }
    }

    public Message receive(List<String> ignoreTypes) throws JMSException {
        for (Entry<String, MessageConsumer> entry : consumers.entrySet()) {
            String type = entry.getKey();
            MessageConsumer consumer = entry.getValue();
            if (ignoreTypes.contains(type))
                continue;
            Message m = consumer.receive();
            if (m != null)
                return m;
        }
        return null;
    }

    public Message receive(long timeout, List<String> ignoreTypes) throws JMSException {
        for (Entry<String, MessageConsumer> entry : consumers.entrySet()) {
            String type = entry.getKey();
            MessageConsumer consumer = entry.getValue();
            if (ignoreTypes.contains(type))
                continue;
            Message m = consumer.receive(timeout / consumers.size());
            if (m != null)
                return m;
        }
        return null;
    }
}
